package org.apache.lucene.ocean.log;

import java.io.IOException;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.ListIterator;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.lucene.ocean.LogDirectory;
import org.apache.lucene.ocean.log.LogFile.Record;
import org.apache.lucene.ocean.util.LongSequence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Manages the log files. There is one active log file at a time that updates
 * are written to. When the active log file reaches the MAX_FILE_SIZE a new
 * active log file is created.
 * 
 */
// TODO: delete log files that are no longer needed
public class LogFileManager {
  final static Logger LOG = LoggerFactory.getLogger(LogFileManager.class);
  private List<LogFile> logFiles = new ArrayList<LogFile>();
  private LongSequence logIdSequence = new LongSequence(1, 1);
  private ReentrantReadWriteLock logFilesLock = new ReentrantReadWriteLock();
  private ScheduledExecutorService logFileCheckTimer;
  private LogDirectory logDirectory;
  private long maxFileSize;
  private long logFileSizeCheckDelayMillis;

  // TODO: load existing max snapshot id
  public LogFileManager(long maxFileSize, long logFileSizeCheckDelayMillis, LogDirectory logDirectory) throws IOException {
    this.logDirectory = logDirectory;
    this.maxFileSize = maxFileSize;
    this.logFileSizeCheckDelayMillis = logFileSizeCheckDelayMillis;
    String[] list = logDirectory.list();
    List<String> logFileNames = new ArrayList<String>(list.length);
    for (String file : list) {
      long id = getLogFileNumber(file);
      LogFile logFile = new LogFile(id, file, logDirectory);
      if (logFiles.size() > 0) {
        if (logFile.getNumRecords() == 0 || logFile.size() == 0) {
          boolean deleted = logFile.delete();
          LOG.info("deleted: " + deleted + " logFile: " + logFile.getName() + " numrecords: " + logFile.getNumRecords());
          // do not add log file that is empty
          // TODO: need to handle log file that has only 1 broken record as it
          // is empty as well
          continue;
        }
      }
      logFiles.add(logFile);
      logFileNames.add(file);
    }
    Collections.sort(logFiles);
    if (logFiles.size() > 0) {
      long last = logFiles.get(logFiles.size() - 1).getId();
      long next = last + 1;
      logIdSequence.set(next);
    }
    LOG.info("initial log files: " + logFileNames);
    logFileCheckTimer = Executors.newSingleThreadScheduledExecutor();
    logFileCheckTimer.scheduleWithFixedDelay(new LogFileSizeCheck(), 1000, logFileSizeCheckDelayMillis, TimeUnit.MILLISECONDS);
  }
  
  public Record getRecord(long id) throws Exception {
    for (LogFile logFile : logFiles) {
      if (logFile.containsId(id)) {
        Record record = logFile.getRecord(id);
        return record;
      }
    }
    return null;
  }
  
  /**
   * Any log file with a maxSnapshotId less than minIndexSnapshotId can be
   * deleted. This is because it means the index would no longer recover
   * anything from the transaction log.
   * 
   * @param minSnapshotId
   */
  public void deleteOldLogFiles(Long minIndexSnapshotId) throws IOException {
    List<LogFile> deletes = new ArrayList<LogFile>();
    logFilesLock.writeLock().lock();
    try {
      Iterator<LogFile> iterator = logFiles.iterator();
      while (iterator.hasNext()) {
        LogFile logFile = iterator.next();
        Long maxId = logFile.getMaxId();
        if (maxId != null) {
          if (minIndexSnapshotId.longValue() > maxId.longValue()) {
            if (LOG.isDebugEnabled())
              LOG.info("minIndexSnapshotId:" + minIndexSnapshotId.longValue() + " maxLogFileId: " + maxId);
            deletes.add(logFile);
            iterator.remove();
          }
        }
      }
    } finally {
      logFilesLock.writeLock().unlock();
    }
    for (LogFile deleteLogFile : deletes) {
      boolean deleted = deleteLogFile.delete();
      if (LOG.isInfoEnabled())
        LOG.info(deleteLogFile.getName() + " deleted: " + deleted);
    }
  }

  public int getNumRecords() {
    logFilesLock.readLock().lock();
    try {
      int num = 0;
      for (LogFile logFile : logFiles) {
        num += logFile.getNumRecords();
      }
      return num;
    } finally {
      logFilesLock.readLock().unlock();
    }
  }

  public void close() throws IOException {
    for (LogFile logFile : logFiles) {
      logFile.close();
    }
  }

  private int getPosition(long id) {
    LogFile lf = new LogFile(id);
    int pos = Collections.binarySearch(logFiles, lf);
    if (pos < 0)
      pos = -1 - pos;
    return pos;
  }

  public Long getMinSnapshotId() {
    if (logFiles.size() > 0) {
      LogFile logFile = logFiles.get(0);
      return logFile.getMinId();
    } else {
      return null;
    }
  }

  public Long getMaxSnapshotId() {
    if (logFiles.size() > 0) {
      LogFile logFile = logFiles.get(logFiles.size() - 1);
      return logFile.getMaxId();
    } else {
      return null;
    }
  }

  public Long getMaxId() {
    if (logFiles.size() == 0)
      return null;
    return logFiles.get(logFiles.size() - 1).getMaxId();
  }

  private LogFile getLast() {
    if (logFiles.size() == 0)
      return null;
    return logFiles.get(logFiles.size() - 1);
  }

  public class LogFileSizeCheck implements Runnable {
    public void run() {
      try {
        LogFile logFile = getLast();
        if (logFile != null) {
          if (logFile.size() >= maxFileSize) {
            LogFile newLogFile = createNewLogFile();
            logFiles.add(newLogFile);
          }
        }
      } catch (IOException ioException) {
        LOG.error("", ioException);
      }
    }
  }

  /**
   * public boolean delete(Long id) throws Exception { LogFile logFile =
   * getLogFileContaining(id); if (logFile == null) throw new Exception("unknown
   * id: " + id); return logFile.delete(id); }
   */
  public Long getPreviousId(Long id) {
    LogFile logFile = getLogFileContaining(id);
    if (logFile == null)
      return null;
    int pos = logFiles.indexOf(logFile);
    Long previousId = logFile.getPreviousId(id);
    if (previousId == null && pos > 0) {
      logFile = logFiles.get(pos - 1);
      previousId = logFile.getPreviousId(id);
    }
    return previousId;
  }

  public long getMinId() {
    if (logFiles.size() > 0) {
      return logFiles.get(0).getMinId();
    }
    return -1;
  }

  public LogFile createNewLogFile() throws IOException {
    long id = logIdSequence.getAndIncrement();
    String fileName = createLogFileName(id);
    if (logDirectory.fileExists(fileName) && logDirectory.fileLength(fileName) > 0) {
      throw new IOException(fileName + " exists");
    }
    LOG.info("creating new log file: " + fileName);
    LogFile logFile = new LogFile(id, fileName, logDirectory);
    logFilesLock.writeLock().lock();
    try {
      logFiles.add(logFile);
      return logFile;
    } finally {
      logFilesLock.writeLock().unlock();
    }
  }

  private static String createLogFileName(long value) {
    DecimalFormat decimalFormat = (DecimalFormat) DecimalFormat.getInstance();
    decimalFormat.applyPattern("00000000");
    return "log" + decimalFormat.format(value) + ".bin";
  }

  private static long getLogFileNumber(String fileName) {
    String numberString = fileName.substring(3, fileName.lastIndexOf('.'));
    return Long.parseLong(numberString);
  }

  private LogFile getLogFileContaining(Long id) {
    logFilesLock.readLock().lock();
    try {
      for (LogFile logFile : logFiles) {
        if (logFile.containsId(id)) {
          return logFile;
        }
      }
      return null;
    } finally {
      logFilesLock.readLock().unlock();
    }
  }

  public boolean contains(Long id) {
    logFilesLock.readLock().lock();
    try {
      for (LogFile logFile : logFiles) {
        if (logFile.containsId(id)) {
          return true;
        }
      }
      return false;
    } finally {
      logFilesLock.readLock().unlock();
    }
  }

  public RecordInfoBackwardIterator getRecordInfosBackwardIterator(Long snapshotId, boolean inclusive) throws Exception {
    return new RecordInfoBackwardIterator(snapshotId, inclusive);
  }

  public class RecordInfoBackwardIterator {
    private ListIterator<LogFile> logFileIterator;
    private ListIterator<RecordHeader> currentIterator;

    public RecordInfoBackwardIterator(Long snapshotId, boolean inclusive) {
      logFileIterator = logFiles.listIterator(logFiles.size()-1);
      while (logFileIterator.hasPrevious()) {
        LogFile logFile = logFileIterator.previous();
        if (snapshotId == null) {
          currentIterator = logFile.recordHeaders.listIterator(logFile.recordHeaders.size()-1);
        } else {
          int index = logFile.getPosition(snapshotId);
          currentIterator = logFile.recordHeaders.listIterator(index);
          if (index >= 0 && !inclusive) previous();
        }
      }
    }

    public RecordHeader previous() {
      if (hasPrevious()) {
        if (currentIterator != null) {
          return currentIterator.previous();
        } else {
          return null;
        }
      } else {
        return null;
      }
    }

    public boolean hasPrevious() {
      if (currentIterator == null)
        return false;
      if (currentIterator.hasPrevious()) {
        return true;
      } else {
        if (logFileIterator.hasPrevious()) {
          LogFile logFile = logFileIterator.previous();
          currentIterator = logFile.recordHeaders.listIterator(logFile.recordHeaders.size()-1);
          return hasPrevious();
        }
      }
      return false;
    }
  }

  public RecordInfoForwardIterator getRecordInfosForwardIterator(Long snapshotId, boolean inclusive) throws Exception {
    return new RecordInfoForwardIterator(snapshotId, inclusive);
  }

  /**
   * RecordHeader forward iterator. Iterates over multiple log files
   * 
   */
  public class RecordInfoForwardIterator {
    private ListIterator<LogFile> logFileIterator;
    private ListIterator<RecordHeader> currentIterator;

    public RecordInfoForwardIterator(Long snapshotId, boolean inclusive) {
      logFileIterator = logFiles.listIterator();
      while (logFileIterator.hasNext()) {
        LogFile logFile = logFileIterator.next();
        if (snapshotId == null) {
          currentIterator = logFile.recordHeaders.listIterator();
          break;
        } else if (logFile.containsId(snapshotId)) {
          int index = logFile.getRawPosition(snapshotId);
          // TODO: need to test this logic
          currentIterator = logFile.recordHeaders.listIterator(index);
          if (index >= 0 && !inclusive) next();
          break;
        }
      }
    }

    public RecordHeader next() {
      if (hasNext()) {
        if (currentIterator != null) {
          return currentIterator.next();
        } else {
          return null;
        }
      } else {
        return null;
      }
    }

    public boolean hasNext() {
      if (currentIterator == null)
        return false;
      if (currentIterator.hasNext()) {
        return true;
      } else {
        if (logFileIterator.hasNext()) {
          LogFile logFile = logFileIterator.next();
          currentIterator = logFile.recordHeaders.listIterator();
          return hasNext();
        }
      }
      return false;
    }
  }

  /**
   * 
   * @param snapshotId
   *          Id from which the iterator starts
   * @return
   * @throws IOException
   */
  public RecordIterator getRecordIterator(Long snapshotId) throws IOException {
    return new RecordIterator(snapshotId);
  }

  public class RecordIterator {
    private Iterator<LogFile> logFileIterator;
    private LogFile.RecordIterator currentRecordIterator;

    public RecordIterator(Long snapshotId) throws IOException {
      logFileIterator = logFiles.iterator();
      while (logFileIterator.hasNext()) {
        LogFile logFile = logFileIterator.next();
        if (snapshotId == null || logFile.containsId(snapshotId)) {
          currentRecordIterator = logFile.getRecordIterator(snapshotId);
        }
      }
    }

    public void close() throws IOException {
      if (currentRecordIterator != null)
        currentRecordIterator.close();
    }

    public Record next() throws IOException {
      if (currentRecordIterator != null) {
        return currentRecordIterator.next();
      } else {
        return null;
      }
    }

    public boolean hasNext() throws IOException {
      if (currentRecordIterator == null)
        return false;
      if (currentRecordIterator.hasNext()) {
        return true;
      } else {
        currentRecordIterator.close();
        if (logFileIterator.hasNext()) {
          LogFile logFile = logFileIterator.next();
          currentRecordIterator = logFile.getRecordIterator(null);
          return hasNext();
        }
      }
      return false;
    }
  }

  public void commit(RecordHeader recordHeader) {
    // TODO: implement commit
  }

  public RecordHeader writeRecord(Long id, RecordData recordData) throws IOException {
    LogFile logFile = getCurrentLogFile();
    return logFile.writeRecord(id, recordData);
  }

  public LogFile getCurrentLogFile() throws IOException {
    logFilesLock.readLock().lock();
    try {
      LogFile logFile = getLast();
      if (logFile == null) {
        logFilesLock.readLock().unlock();
        logFile = createNewLogFile();
        logFilesLock.readLock().lock();
      }
      return logFile;
    } finally {
      logFilesLock.readLock().unlock();
    }
  }
}
